package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseAppV2;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.bean.PaymentWide;
import com.atguigu.gmall.realtime.bean.ProductStats;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.function.DimAsyncFunction;
import com.atguigu.gmall.realtime.util.AtguiguUtil;
import com.atguigu.gmall.realtime.util.DimUtil;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static com.atguigu.gmall.realtime.common.Constant.*;

/**
 * @author Archie
 * @date 2021-10-27 14:37
 * @description
 */
public class DwsProductStatsApp extends BaseAppV2 {

	public static void main(String[] args) {
		new DwsProductStatsApp().init(
				4002,
				1,
				"DwsProductStatsApp",
				"DwsProductStatsApp",
				TOPIC_DWD_PAGE, TOPIC_DWD_DISPLAY,
				TOPIC_DWM_ORDER_WIDE,TOPIC_DWM_PAYMENT_WIDE,
				TOPIC_DWD_FAVOR_INFO, TOPIC_DWD_CART_INFO,
				TOPIC_DWD_REFUND_PAYMENT, TOPIC_DWD_COMMENT_INFO
		);
	}

	@Override
	protected void run(StreamExecutionEnvironment env, Map<String, DataStreamSource<String>> streams) {
		
		// 1. 解析并Union
		DataStream<ProductStats> psStream = parseAndUnionOne(streams);
		// 2. 开窗聚合
		SingleOutputStreamOperator<ProductStats> aggregatedStream = aggregate(psStream);
		// 3. join 维度（redis缓存 + 异步发送）
		SingleOutputStreamOperator<ProductStats> psStreamWithDim = joinDim(aggregatedStream);
		// 4. 写入 CK
		writeToClickHouse(psStreamWithDim);
		// 5. 由于商品行为关键词需要, 把数据另外写一份到kafka中
		writeToKafka(psStreamWithDim);

	}

	private void writeToKafka(SingleOutputStreamOperator<ProductStats> stream) {
		stream
				.map(JSON::toJSONString)
				.addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWS_PRODUCT_STATS));
	}

	private void writeToClickHouse(SingleOutputStreamOperator<ProductStats> psStreamWithDim) {
		psStreamWithDim.addSink(FlinkSinkUtil.getClickHouseSink(
				"gmall2021", "product_stats_2021", ProductStats.class
		));
	}

	private SingleOutputStreamOperator<ProductStats> joinDim(SingleOutputStreamOperator<ProductStats> aggregatedStream) {
		return AsyncDataStream.unorderedWait(
				aggregatedStream,
				new DimAsyncFunction<ProductStats>() {
					@Override
					public void addDim(Connection phoenixConn,
									   Jedis client,
									   ProductStats ps,
									   ResultFuture<ProductStats> resultFuture) {

						JSONObject skuInfo = DimUtil.getDim(client, phoenixConn, "dim_sku_info", ps.getSku_id());

						// >>> 补齐SkuInfo
						ps.setSku_name(skuInfo.getString("SKU_NAME"));
						// >>> 补齐 spu_id / tm_id / c3_id
						ps.setSku_price(skuInfo.getBigDecimal("PRICE"));
						ps.setSpu_id(skuInfo.getLong("SPU_ID"));
						ps.setTm_id(skuInfo.getLong("TM_ID"));
						ps.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
						// >>> spu
						JSONObject spuInfo = DimUtil.getDim(client, phoenixConn, "dim_spu_info", ps.getSpu_id());
						ps.setSpu_name(spuInfo.getString("SPU_NAME"));
						// >>> tm
						JSONObject tmInfo = DimUtil.getDim(client, phoenixConn, "dim_base_trademark", ps.getTm_id());
						ps.setTm_name(tmInfo.getString("TM_NAME"));
						// >>> c3
						JSONObject c3Info = DimUtil.getDim(client, phoenixConn, "dim_base_category3", ps.getCategory3_id());
						ps.setCategory3_name(c3Info.getString("NAME"));

						// <<< 数据放入 resultFuture 中
						resultFuture.complete(Collections.singletonList(ps));
					}
				},
				60,
				TimeUnit.SECONDS
		);
	}

	private SingleOutputStreamOperator<ProductStats> aggregate(DataStream<ProductStats> psStream) {

		return psStream
				.assignTimestampsAndWatermarks(
						WatermarkStrategy
								.<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(5))
						.withTimestampAssigner((data, ts) -> data.getTs())
				)
				.keyBy(ProductStats::getSku_id)
				.window(TumblingEventTimeWindows.of(Time.seconds(5)))
				.reduce(
						new ReduceFunction<ProductStats>() {
							@Override
							public ProductStats reduce(ProductStats ps1, ProductStats ps2) throws Exception {
								ps1.setClick_ct(ps1.getClick_ct() + ps2.getClick_ct());
								ps1.setDisplay_ct(ps1.getDisplay_ct() + ps2.getDisplay_ct());
								ps1.setFavor_ct(ps1.getFavor_ct() + ps2.getFavor_ct());
								ps1.setCart_ct(ps1.getCart_ct() + ps2.getCart_ct());

								ps1.setOrder_amount(ps1.getOrder_amount().add(ps2.getOrder_amount()));
								ps1.setPayment_amount(ps1.getPayment_amount().add(ps2.getPayment_amount()));
								ps1.setRefund_amount(ps1.getRefund_amount().add(ps2.getRefund_amount()));

								ps1.setOrder_sku_num(ps1.getOrder_sku_num() + ps2.getOrder_sku_num());
								ps1.setComment_ct(ps1.getComment_ct() + ps2.getComment_ct());
								ps1.setGood_comment_ct(ps1.getGood_comment_ct() + ps2.getGood_comment_ct());

								ps1.getOrderIdSet().addAll(ps2.getOrderIdSet());
								ps1.getRefundOrderIdSet().addAll(ps2.getRefundOrderIdSet());
								ps1.getPaidOrderIdSet().addAll(ps2.getPaidOrderIdSet());

								return ps1;
							}
						},
						new ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
							@Override
							public void process(Long aLong, Context context, Iterable<ProductStats> elements, Collector<ProductStats> out) throws Exception {
								ProductStats ps = elements.iterator().next();
								TimeWindow w = context.window();
								ps.setStt(AtguiguUtil.toDataTimeString(w.getStart()));
								ps.setEdt(AtguiguUtil.toDataTimeString(w.getEnd()));
								ps.setTs(System.currentTimeMillis());
								ps.setOrder_ct((long) ps.getOrderIdSet().size());
								ps.setRefund_order_ct((long) ps.getRefundOrderIdSet().size());
								ps.setPaid_order_ct((long) ps.getPaidOrderIdSet().size());

								out.collect(ps);
							}
						}
				);
	}

	private DataStream<ProductStats> parseAndUnionOne(Map<String, DataStreamSource<String>> streams) {

		/* >>> 点击信息汇聚 <<< */
		SingleOutputStreamOperator<ProductStats> clickStream = streams
				.get(TOPIC_DWD_PAGE)
				.flatMap(new FlatMapFunction<String, ProductStats>() {
					@Override
					public void flatMap(String value, Collector<ProductStats> out) throws Exception {
						JSONObject obj = JSON.parseObject(value);
						JSONObject page = obj.getJSONObject("page");
						String pageId = page.getString("page_id");
						String itemType = page.getString("item_type");

						if ("good_detail".equals(pageId) && "sku_id".equals(itemType)) { // 过滤出和商品相关的点击信息
							Long skuId = page.getLong("item");
							ProductStats productStats = new ProductStats();
							productStats.setSku_id(skuId); // 取商品id
							productStats.setClick_ct(1L); // 取点击量(默认置1)
							productStats.setTs(obj.getLong("ts")); // 取时间戳
							out.collect(productStats);
						}
					}
				});

		/* >>> 曝光信息汇聚 <<< */
		SingleOutputStreamOperator<ProductStats> displayStream = streams
				.get(TOPIC_DWD_DISPLAY)
				.filter(value -> {
					JSONObject obj = JSON.parseObject(value);
					String item = obj.getString("item");
					return item.matches("^\\d+$"); // ===> TODO...
				})
				.map(value -> {
					JSONObject obj = JSON.parseObject(value);
					ProductStats productStats = new ProductStats();
					productStats.setSku_id(obj.getLong("item")); // 取商品id
					productStats.setDisplay_ct(1L); // 取曝光度（默认置1）
					productStats.setTs(obj.getLong("ts")); // 取时间戳
					return productStats;
				});

		/* >>> 订单信息汇聚 <<< */
		SingleOutputStreamOperator<ProductStats> orderStream = streams
				.get(TOPIC_DWM_ORDER_WIDE)
				.map(value -> {
					OrderWide ow = JSON.parseObject(value, OrderWide.class);
					ProductStats productStats = new ProductStats();
					productStats.setSku_id(ow.getSku_id()); // 取商品id
					productStats.getOrderIdSet().add(ow.getOrder_id()); // 累计id集合（已去重）
					productStats.setOrder_amount(ow.getSplit_total_amount()); // 取 订单总数
					productStats.setOrder_sku_num(ow.getSku_num()); // 取 下单商品数目
					productStats.setTs(AtguiguUtil.toTs(ow.getCreate_time())); // 取时间戳
					return productStats;
				});

		/* >>>  支付信息汇聚 <<< */
		SingleOutputStreamOperator<ProductStats> paymentStream = streams
				.get(TOPIC_DWM_PAYMENT_WIDE)
				.map(value -> {
					PaymentWide pw = JSON.parseObject(value, PaymentWide.class);
					ProductStats productStats = new ProductStats();
					productStats.setSku_id(pw.getSku_id());
					productStats.getPaidOrderIdSet().add(pw.getOrder_id());
					productStats.setPayment_amount(pw.getSplit_total_amount());
					productStats.setTs(AtguiguUtil.toTs(pw.getPayment_create_time()));
					return productStats;
				});

		/* >>> 收藏信息汇聚 <<< */
		SingleOutputStreamOperator<ProductStats> favorStream = streams
				.get(TOPIC_DWD_FAVOR_INFO)
				.map(value -> {
					JSONObject obj = JSON.parseObject(value);
					ProductStats ps = new ProductStats();
					ps.setSku_id(obj.getLong("sku_id"));
					ps.setTs(AtguiguUtil.toTs(obj.getString("create_time")));
					ps.setFavor_ct(1L);
					return ps;
				});

		/* >>> 购物车信息汇聚 <<< */
		SingleOutputStreamOperator<ProductStats> cartStream = streams
				.get(TOPIC_DWD_CART_INFO)
				.map(value -> {
					JSONObject obj = JSON.parseObject(value);
					ProductStats ps = new ProductStats();
					ps.setSku_id(obj.getLong("sku_id"));
					ps.setTs(AtguiguUtil.toTs(obj.getString("create_time")));
					ps.setCart_ct(1L);
					return ps;
				});

		/* >>> 退款信息汇聚 <<< */
		SingleOutputStreamOperator<ProductStats> refundStream = streams
				.get(TOPIC_DWD_REFUND_PAYMENT)
				.map(value -> {
					JSONObject refund = JSON.parseObject(value);
					ProductStats ps = new ProductStats();
					ps.setSku_id(refund.getLong("sku_id"));
					ps.getRefundOrderIdSet().add(refund.getLong("order_id"));
					ps.setRefund_amount(refund.getBigDecimal("total_amount"));
					ps.setTs(AtguiguUtil.toTs(refund.getString("create_time")));
					return ps;
				});

		/* >>> 评价信息汇聚 <<< */
		SingleOutputStreamOperator<ProductStats> commentStream = streams
				.get(TOPIC_DWD_COMMENT_INFO)
				.map(value -> {
					JSONObject refund = JSON.parseObject(value);
					ProductStats ps = new ProductStats();
					ps.setSku_id(refund.getLong("sku_id")); // 取商品id
					ps.setTs(AtguiguUtil.toTs(refund.getString("create_time"))); // 取时间戳
					ps.setComment_ct(1L);
					// 判断是否好评
					String appraise = refund.getString("appraise");
					if (FIVE_STAR.equals(appraise) || FOUR_STAR.equals(appraise)) {
						ps.setGood_comment_ct(1L); // 取好评单数
					}
					return ps;
				});

		return clickStream.union(
				displayStream,
				favorStream,
				cartStream,
				orderStream,
				paymentStream,
				refundStream,
				commentStream);
	}
}
